package com.dtp.core.support;

import com.alibaba.ttl.TtlRunnable;
import com.alibaba.ttl.threadpool.TtlExecutors;
import com.dtp.common.constant.DynamicTpConst;
import com.dtp.common.entity.NotifyItem;
import com.dtp.common.em.NotifyItemEnum;
import com.dtp.common.em.QueueTypeEnum;
import com.dtp.common.em.RejectedTypeEnum;
import com.dtp.common.queue.VariableLinkedBlockingQueue;
import com.dtp.core.reject.RejectHandlerGetter;
import com.dtp.core.support.wrapper.TaskWrapper;
import com.dtp.core.thread.DtpExecutor;
import com.dtp.core.thread.ScheduledDtpExecutor;
import com.dtp.core.thread.EagerDtpExecutor;
import com.dtp.core.thread.NamedThreadFactory;
import com.dtp.core.thread.OrderedDtpExecutor;
import com.google.common.collect.Lists;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.Assert;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import cn.hutool.core.lang.func.Func1;

/**
 * 生成器，用于优雅地创建ThreadPoolExecutor
 *
 * @author yanhom
 * @since 1.0.0
 *
 * @remark muse
 **/
public class ThreadPoolBuilder {
    private String threadPoolName = "DynamicTp";    // 动态线程池名称
    private int corePoolSize = 1;                   // 核心线程数大小
    private int maximumPoolSize = DynamicTpConst.AVAILABLE_PROCESSORS; // 最大线程数大小
    private long keepAliveTime = 60;                // 当线程数量大于核心时，这是多余空闲线程在终止前等待新任务的最大时间
    private TimeUnit timeUnit = TimeUnit.SECONDS;   // 超时单位
    private BlockingQueue<Runnable> workQueue = new VariableLinkedBlockingQueue<>(1024); // 阻塞队列{@link QueueTypeEnum}
    private int queueCapacity = 1024;   // 队列容量
    private int maxFreeMemory = 256;    // MemorySafeLBQ的最大剩余内存，单位M
    private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy(); // 拒绝策略{@link RejectedTypeEnum}
    private ThreadFactory threadFactory = new NamedThreadFactory("dtp");    // 默认的内部线程工厂
    private boolean allowCoreThreadTimeOut = false;                         // 是否允许核心现成超时
    private boolean dynamic = true;                                         // 动态线程池 还是 普通线程池
    private boolean waitForTasksToCompleteOnShutdown = false;   // 是否在关机时等待计划任务完成，不中断正在运行的任务，执行队列中的所有任务
    private int awaitTerminationSeconds = 0;        // 在容器的其他部分继续关闭之前，为了等待剩余的任务完成执行，执行器应该阻塞的最大秒数
    private boolean ioIntensive = false;            // 如果io密集型线程池。默认为false, true表示CPU密集型线程池。
    private boolean ordered = false;                // 如果有序线程池。默认为false, true有序线程池。
    private boolean scheduled = false;              // 如果是计划线程池，默认为false。
    private boolean preStartAllCoreThreads = false; // 是否提前开启所有核心线程
    private long runTimeout = 0;    // 任务执行超时时间，单位(毫秒)，仅用于统计。
    private long queueTimeout = 0;  // 任务队列等待超时时间，单位(毫秒)，仅供统计。
    private final List<TaskWrapper> taskWrappers = Lists.newArrayList();    // 任务包装器集合
    private List<NotifyItem> notifyItems = NotifyItem.getAllNotifyItems();  // 通知项{@link NotifyItemEnum}
    private List<String> platformIds = Lists.newArrayList();                // 通知平台id

    private ThreadPoolBuilder() { }

    /** 创建阻塞队列 */
    public ThreadPoolBuilder workQueue(String queueName, Integer capacity, Boolean fair, Integer maxFreeMemory) {
        if (StringUtils.isNotBlank(queueName)) {
            workQueue = QueueTypeEnum.buildLbq(queueName, capacity != null ? capacity : this.queueCapacity,
                    fair != null && fair, maxFreeMemory != null ? maxFreeMemory : this.maxFreeMemory);
        }
        return this;
    }

    /** 创建阻塞队列 */
    public ThreadPoolBuilder workQueue(String queueName, Integer capacity, Boolean fair) {
        if (StringUtils.isNotBlank(queueName)) {
            workQueue = QueueTypeEnum.buildLbq(queueName, capacity != null ? capacity : this.queueCapacity,
                    fair != null && fair, maxFreeMemory);
        }
        return this;
    }

    /** 根据dynamic参数，构建线程池（true：构建DtpExecutor  false：构建ThreadPoolExecutor）*/
    public ThreadPoolExecutor build() {
        if (dynamic) return buildDtpExecutor(this);
        else return buildCommonExecutor(this);
    }

    /** 构建动态线程池*/
    public DtpExecutor buildDynamic() {return buildDtpExecutor(this);}

    /** 构建普通线程池 */
    public ThreadPoolExecutor buildCommon() {return buildCommonExecutor(this);}

    /** 使用ttl（transmittable-thread-local）构建线程池和封装器 */
    public ExecutorService buildWithTtl() {
        if (dynamic) {
            taskWrappers.add(TtlRunnable::get);
            return buildDtpExecutor(this);
        } else {
            return TtlExecutors.getTtlExecutorService(buildCommonExecutor(this));
        }
    }

    /** 构建动态线程池——DtpExecutor */
    private DtpExecutor buildDtpExecutor(ThreadPoolBuilder builder) {
        Assert.notNull(builder.threadPoolName, "The thread pool name must not be null.");
        DtpExecutor dtpExecutor = createInternal(builder);
        dtpExecutor.setThreadPoolName(builder.threadPoolName);
        dtpExecutor.allowCoreThreadTimeOut(builder.allowCoreThreadTimeOut);
        dtpExecutor.setWaitForTasksToCompleteOnShutdown(builder.waitForTasksToCompleteOnShutdown);
        dtpExecutor.setAwaitTerminationSeconds(builder.awaitTerminationSeconds);
        dtpExecutor.setPreStartAllCoreThreads(builder.preStartAllCoreThreads);
        dtpExecutor.setRunTimeout(builder.runTimeout);
        dtpExecutor.setQueueTimeout(builder.queueTimeout);
        dtpExecutor.setTaskWrappers(builder.taskWrappers);
        dtpExecutor.setNotifyItems(builder.notifyItems);
        dtpExecutor.setPlatformIds(builder.platformIds);
        return dtpExecutor;
    }
    private DtpExecutor createInternal(ThreadPoolBuilder builder) {
        DtpExecutor dtpExecutor;
        if (ioIntensive) { // io密集型线程池
            TaskQueue taskQueue = new TaskQueue(builder.queueCapacity);
            dtpExecutor = new EagerDtpExecutor(
                    builder.corePoolSize,
                    builder.maximumPoolSize,
                    builder.keepAliveTime,
                    builder.timeUnit,
                    taskQueue,
                    builder.threadFactory,
                    builder.rejectedExecutionHandler);
            taskQueue.setExecutor((EagerDtpExecutor) dtpExecutor);
        } else if (ordered) { // 有序线程池
            dtpExecutor = new OrderedDtpExecutor(
                    builder.corePoolSize,
                    builder.maximumPoolSize,
                    builder.keepAliveTime,
                    builder.timeUnit,
                    builder.workQueue,
                    builder.threadFactory,
                    builder.rejectedExecutionHandler);
        } else if (scheduled) { // 计划线程池
            dtpExecutor = new ScheduledDtpExecutor(
                    builder.corePoolSize,
                    builder.maximumPoolSize,
                    builder.keepAliveTime,
                    builder.timeUnit,
                    builder.workQueue,
                    builder.threadFactory,
                    builder.rejectedExecutionHandler);
        } else { // 普通动态线程池
            dtpExecutor = new DtpExecutor(
                    builder.corePoolSize,
                    builder.maximumPoolSize,
                    builder.keepAliveTime,
                    builder.timeUnit,
                    builder.workQueue,
                    builder.threadFactory,
                    builder.rejectedExecutionHandler);
        }
        return dtpExecutor;
    }

    /** 构建公共threadPoolExecutor，不被DynamicTp框架管理 */
    private ThreadPoolExecutor buildCommonExecutor(ThreadPoolBuilder builder) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                builder.corePoolSize,
                builder.maximumPoolSize,
                builder.keepAliveTime,
                builder.timeUnit,
                builder.workQueue,
                builder.threadFactory,
                builder.rejectedExecutionHandler
        );
        executor.allowCoreThreadTimeOut(builder.allowCoreThreadTimeOut);
        return executor;
    }

    /** ---------------------如下是为构建者模式提供的方法--------------------- */
    public static ThreadPoolBuilder newBuilder() {return new ThreadPoolBuilder();}
    public ThreadPoolBuilder threadPoolName(String poolName) {
        this.threadPoolName = poolName;
        return this;
    }
    public ThreadPoolBuilder corePoolSize(int corePoolSize) {
        if (corePoolSize >= 0) this.corePoolSize = corePoolSize;
        return this;
    }
    public ThreadPoolBuilder maximumPoolSize(int maximumPoolSize) {
        if (maximumPoolSize > 0) this.maximumPoolSize = maximumPoolSize;
        return this;
    }
    public ThreadPoolBuilder keepAliveTime(long keepAliveTime) {
        if (keepAliveTime > 0) this.keepAliveTime = keepAliveTime;
        return this;
    }
    public ThreadPoolBuilder timeUnit(TimeUnit timeUnit) {
        if (timeUnit != null) this.timeUnit = timeUnit;
        return this;
    }
    public ThreadPoolBuilder queueCapacity(int queueCapacity) {
        this.queueCapacity = queueCapacity;
        return this;
    }
    public ThreadPoolBuilder maxFreeMemory(int maxFreeMemory) {
        this.maxFreeMemory = maxFreeMemory;
        return this;
    }
    public ThreadPoolBuilder rejectedExecutionHandler(String rejectedName) {
        if (StringUtils.isNotBlank(rejectedName)) rejectedExecutionHandler = RejectHandlerGetter.buildRejectedHandler(rejectedName);
        return this;
    }
    public ThreadPoolBuilder threadFactory(String prefix) {
        if (StringUtils.isNotBlank(prefix)) threadFactory = new NamedThreadFactory(prefix);
        return this;
    }
    public ThreadPoolBuilder allowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
        this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
        return this;
    }
    public ThreadPoolBuilder dynamic(boolean dynamic) {
        this.dynamic = dynamic;
        return this;
    }
    public ThreadPoolBuilder awaitTerminationSeconds(int awaitTerminationSeconds) {
        this.awaitTerminationSeconds = awaitTerminationSeconds;
        return this;
    }
    public ThreadPoolBuilder waitForTasksToCompleteOnShutdown(boolean waitForTasksToCompleteOnShutdown) {
        this.waitForTasksToCompleteOnShutdown = waitForTasksToCompleteOnShutdown;
        return this;
    }
    public ThreadPoolBuilder ioIntensive(boolean ioIntensive) {
        this.ioIntensive = ioIntensive;
        return this;
    }
    public ThreadPoolBuilder ordered(boolean ordered) {
        this.ordered = ordered;
        return this;
    }
    public ThreadPoolBuilder scheduled(boolean scheduled) {
        this.scheduled = scheduled;
        return this;
    }
    public ThreadPoolBuilder preStartAllCoreThreads(boolean preStartAllCoreThreads) {
        this.preStartAllCoreThreads = preStartAllCoreThreads;
        return this;
    }
    public ThreadPoolBuilder runTimeout(long runTimeout) {
        this.runTimeout = runTimeout;
        return this;
    }
    public ThreadPoolBuilder queueTimeout(long queueTimeout) {
        this.queueTimeout = queueTimeout;
        return this;
    }
    public ThreadPoolBuilder taskWrappers(List<TaskWrapper> taskWrappers) {
        this.taskWrappers.addAll(taskWrappers);
        return this;
    }
    public ThreadPoolBuilder taskWrapper(TaskWrapper taskWrapper) {
        this.taskWrappers.add(taskWrapper);
        return this;
    }
    public ThreadPoolBuilder notifyItems(List<NotifyItem> notifyItemList) {
        if (CollectionUtils.isNotEmpty(notifyItemList)) notifyItems = notifyItemList;
        return this;
    }

    public ThreadPoolBuilder platformIds(List<String> platformIds) {
        if (CollectionUtils.isNotEmpty(platformIds)) this.platformIds = platformIds;
        return this;
    }
}
